package com.burges.net.dataStream.waterMark

import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.watermark.Watermark

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/1/27 
  * 描述     自定义实现Punctuated Watermark
  */
class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[(String, Long, Int)]{

	//复写checkAndGetNextWatermark方法定义Watermark生成逻辑
	override def checkAndGetNextWatermark(element : (String, Long, Int), l: Long): Watermark = {
		element._2
	}

	//复写extractTimestamp方法定义抽取timestamp逻辑
	override def extractTimestamp(lastElement: (String, Long, Int), extractedTimestamp: Long): Watermark = {
		if(lastElement._3 == 0)
			new Watermark(extractedTimestamp)
		else
			null
	}

}
